package com.gandong8.hbase.tool.service.impl;

import com.gandong8.hbase.tool.annotation.HbaseColumn;
import com.gandong8.hbase.tool.annotation.HbaseTable;
import com.gandong8.hbase.tool.domain.*;
import com.gandong8.hbase.tool.service.HbaseServiceInter;
import com.gandong8.hbase.tool.util.HbaseUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.*;
import java.util.stream.Collectors;

/**
 * HbaseServiceImpl
 * hbase常用操作实现类
 * @author xiaoyutou
 */
@Service
@Slf4j
public class HbaseServiceImpl implements HbaseServiceInter {

    @Autowired
    Connection hbaseConnection;

    @Value("${hbase.web.query.maxrow}")
    private int mymaxrow;

    private HashMap<Class<?>, String> tableMap = new HashMap<>();


    
    @Override
    public boolean createTable(String tableName, Set<String> familySet) {
        log.info("begin_create:{},{}", tableName, familySet);
        try(Admin admin = getAdmin())  {
            HTableDescriptor tableDescriptor = new HTableDescriptor(getHbaseTableName(tableName));
            familySet.forEach((c)->{
                tableDescriptor.addFamily(new HColumnDescriptor(c));
            });
            Objects.requireNonNull(admin).createTable(tableDescriptor);
            return true;
        } catch (TableExistsException ee) {
            log.error("TableExistsException:{}", tableName);
        } catch (Exception e) {
            log.error("createTable_err", e);
        }
        return false;
    }

    @Override
    public boolean createTable(Class<?> clazz) {
        try {
            Field[] fields = clazz.getDeclaredFields();
            Set<String> set = new HashSet<>();
            for (Field field : fields) {
                HbaseColumn column = field.getAnnotation(HbaseColumn.class);
                if (column == null) {
                    continue;
                }
                field.setAccessible(true);
                String family = column.family();
                if (ConstantDO.ROW_KEY.equalsIgnoreCase(family)) {
                    continue;
                }
                set.add(family);
            }
            createTable(getTableName(clazz), set);
            return true;
        } catch (Exception e){
            log.error("createTable_err:{}", clazz, e);
        }
        return false;
    }

    @Override
    public boolean deleteTable(String tableName) {
        log.info("begin_delete:{}", tableName);
        TableName tableName1 = TableName.valueOf(tableName);
        try(Admin admin = getAdmin())  {
            Objects.requireNonNull(admin).disableTable(tableName1);
            admin.deleteTable(tableName1);
            return true;
        } catch (TableNotFoundException ee) {
            log.error("TableNotFoundException:{}", tableName);
        } catch (Exception e) {
            log.error("deleteTable_err", e);
        }
        return false;
    }

    @Override
    public List<String> listTable() {
        List<String> tables = new ArrayList<>();
        try(Admin admin = getAdmin())  {
            TableName[] tableNames = Objects.requireNonNull(admin).listTableNames();
            for (TableName tableName : tableNames) {
                tables.add(tableName.getNameAsString());
            }
        } catch (Exception e) {
            log.error("listTable_err", e);
        }
        return tables;
    }

    @Override
    public List<HbaseTableInfo> listTableInfo() {
        List<HbaseTableInfo> tables = new ArrayList<>();
        try(Admin admin = getAdmin())  {
            TableName[] tableNames = Objects.requireNonNull(admin).listTableNames();
            for (TableName tableName : tableNames) {
                HbaseTableInfo tableInfo = new HbaseTableInfo();
                tableInfo.setName(tableName.getNameAsString());
                tableInfo.setIsTableEnabled(admin.isTableEnabled(tableName));
                tableInfo.setFamilies(listFamily(tableInfo.getName()));
                tables.add(tableInfo);
            }
        } catch (Exception e) {
            log.error("listTableInfo_err", e);
        }
        return tables;
    }

    @Override
    public HbaseClusterInfo getClusterInfo() {
        try(Admin admin = getAdmin())  {
            if(admin == null) {
                return null;
            }
            ClusterStatus status = admin.getClusterStatus();
            HbaseClusterInfo info = new HbaseClusterInfo();
            Collection<ServerName> serverNames = status.getServers();
            List<String> serverList = serverNames.stream().map(ServerName::getServerName).collect(Collectors.toList());
            info.setServerList(serverList);
            info.setClusterID(status.getClusterId());
            info.setHbaseVersion(status.getHBaseVersion());
            info.setRegionsCount(status.getRegionsCount());
            return info;
        } catch (Exception e) {
            log.error("getClusterStatus_err", e);
        }
        return null;
    }

    private void printHbaseInfo() {
        try(Admin admin = getAdmin())  {
            assert admin != null;
            ClusterStatus status = admin.getClusterStatus();
            //print cluster status
            System.out.println("HBaseVersion: " + status.getHBaseVersion());
            System.out.println("Version: "+status.getVersion());
            System.out.println("ServerSize: " + status.getServersSize());
            System.out.println("ClusterID: " + status.getClusterId());
            System.out.println("Servers: "+ status.getServers());
            System.out.println("DeadServer: " + status.getDeadServerNames());
            System.out.println("RegionsCount: " + status.getRegionsCount());
            System.out.println("Regions in Transition: " + status.getRegionsInTransition());
            System.out.println("RequestsCount: " + status.getRequestsCount());
            System.out.println("AverageLoad: "+status.getAverageLoad());
            for(ServerName server:status.getServers()) {
                //print server info
                System.out.println("Hostname: " + server.getHostname());
                System.out.println("Host and Port: " + server.getHostAndPort());
                System.out.println("ServerName: " + server.getServerName());
                System.out.println("Port: " + server.getPort());
                System.out.println("StartCode: " + server.getStartcode());
                ServerLoad load = status.getLoad(server);
                System.out.println("Load: " + load.getLoad());
                System.out.println("MaxHeap(MB): " + load.getMaxHeapMB());
                System.out.println("MemStoreSize(MB): " + load.getMemstoreSizeInMB());
                System.out.println("NumberOfRegions: " + load.getNumberOfRegions());
                System.out.println("NumberOfRequests: " + load.getNumberOfRequests());
                System.out.println("StoreFileIndexSize(MB): " + load.getStorefileIndexSizeInMB());
                System.out.println("StoreFiles: " + load.getStorefiles());
                System.out.println("StoreFileSize(MB): " + load.getStorefileSizeInMB());
                System.out.println("UsedHeap(MB): " + load.getUsedHeapMB());
                //print region info
                for (Map.Entry<byte[], RegionLoad> entry : load.getRegionsLoad().entrySet()) {
                    System.out.println("Region: " + Bytes.toStringBinary(entry.getKey()));
                    RegionLoad regionLoad = entry.getValue();
                    System.out.println("Name: " + Bytes.toStringBinary(regionLoad.getName()));
                    System.out.println("Stores: " + regionLoad.getStores());
                    System.out.println("StoreFiles: " + regionLoad.getStorefiles());
                    System.out.println("StoreFileSize(MB): " + regionLoad.getStorefileSizeMB());
                    System.out.println("StoreFileIndexSize(MB): " + regionLoad.getStorefileIndexSizeMB());
                    System.out.println("MemStoreSize(MB): " + regionLoad.getMemStoreSizeMB());
                    System.out.println("RequestCount: " + regionLoad.getRequestsCount());
                    System.out.println("ReadRequestsCount: " + regionLoad.getReadRequestsCount());
                    System.out.println("WriteRequestsCount: " + regionLoad.getWriteRequestsCount());
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    @Override
    public boolean enableTable(String tableName) {
        try(Admin admin = getAdmin())  {
            Objects.requireNonNull(admin).enableTable(getHbaseTableName(tableName));
           return true;
        } catch (Exception e) {
            log.error("enableTable_err", e);
        }
        return false;
    }

    @Override
    public boolean disableTable(String tableName) {
        try(Admin admin = getAdmin())  {
            Objects.requireNonNull(admin).disableTable(getHbaseTableName(tableName));
            return true;
        } catch (Exception e) {
            log.error("disableTable_err", e);
        }
        return false;
    }

    @Override
    public List<String> listFamily(String tableName) {
        List<String> columnList = new ArrayList<>();
        try(Table table = getHbaseTable(tableName)) {
            HColumnDescriptor[] columnDescriptors = table.getTableDescriptor().getColumnFamilies();
            for (HColumnDescriptor hColumnDescriptor :columnDescriptors) {
                columnList.add(hColumnDescriptor.getNameAsString());
            }
            return columnList;
        } catch (Exception e) {
            log.error("listColumn_err:{}" ,tableName, e);
        }
        return columnList;
    }

    @Override
    public  List<String> listFamily(Class<?> clazz) {
        return listFamily(getTableName(clazz));
    }

    /**
     * 删除多行
     * @param clazz
     * @param rowKeys
     */
    @Override
    public boolean delete(Class<?> clazz, List<String> rowKeys) {
        return delete(getTableName(clazz), rowKeys);
    }

    @Override
    public boolean delete(String tableName, List<String> rowKeys) {
        List<Delete> deletes = rowKeys.stream().map(
                rowKey -> new Delete(Bytes.toBytes(rowKey))).collect(Collectors.toList()
        );
        try(Table table = getHbaseTable(tableName)) {
            table.delete(deletes);
            return true;
        } catch (IOException e) {
            log.error("delete_rows_err:{}", rowKeys, e);
        }
        return false;
    }

    /**
     * 删除行
     * @param clazz
     * @param rowKey
     */
    @Override
    public boolean delete(Class<?> clazz, String rowKey) {
        return delete(getTableName(clazz), rowKey);
    }

    /**
     * 删除行
     * @param tableName
     * @param rowKey
     */
    @Override
    public boolean delete(String tableName, String rowKey) {
        Delete del = new Delete(Bytes.toBytes(rowKey));
        try(Table table = getHbaseTable(tableName))  {
            table.delete(del);
            return true;
        } catch (IOException e) {
            log.error("delete_row_err:{}", rowKey, e);
        }
        return false;
    }

    @Override
    public boolean deleteColumn(String tableName, String rowKey, String family, String qualifier) {
        Delete del = new Delete(Bytes.toBytes(rowKey));
        del.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        try(Table table = getHbaseTable(tableName))  {
            table.delete(del);
            return true;
        } catch (IOException e) {
            log.error("delete_column_err:{}", rowKey, e);
        }
        return false;
    }

    /**
     * 删除行
     * @param obj
     * @param <T>
     */
    @Override
    public <T> void delete(T obj) {
        Delete del = new Delete(Bytes.toBytes(HbaseUtil.getRowKey(obj)));
        try(Table table = getHbaseTable(obj.getClass()))  {
            table.delete(del);
        } catch (IOException e) {
            log.error("delete_obj_err:{}", obj, e);
        }
    }


    private TableName getHbaseTableName(String tableName) {
        return TableName.valueOf(tableName);
    }

    private String getTableName(Class<?> clazz) {
        if(tableMap.containsKey(clazz)) {
            return tableMap.get(clazz);
        }
        HbaseTable table = clazz.getAnnotation(HbaseTable.class);
        tableMap.put(clazz, table.name());
        return table.name();
    }

    @Override
    public  <T> boolean save(T obj) {
        List<Put> puts = new ArrayList<>();
        try {
            Put put = HbaseUtil.objectToPut(obj);
            puts.add(put);
        }catch (Exception e){
            log.error("save_err:{}", obj, e);
        }
        return savePut(puts, obj.getClass());
    }

    @Override
    public  <T> boolean save(List<T> objList) {
        List<Put> puts = new ArrayList<>();
        try {
            for (Object obj : objList) {
                Put put = HbaseUtil.objectToPut(obj);
                puts.add(put);
            }
        } catch (Exception e){
            log.error("saveList_err:{}", objList, e);
        }
        return savePut(puts, objList.get(0).getClass());
    }

    @Override
    public boolean save(String tableName, String rowKey, String[] columns, String[] values) {
        if(tableName == null || rowKey == null || columns == null || values == null) {
            return false;
        }
        int len = columns.length;
        if(len < 1 || len != values.length) {
            return false;
        }
        List<HbaseData> dataList = new ArrayList<>();
        for (int i = 0; i < len; i++) {
            HbaseData hbaseData = parseColumnData(columns[i], values[i]);
            if(hbaseData == null) {
                continue;
            }
            dataList.add(hbaseData);
        }
        List<Put> puts = new ArrayList<>();
        puts.add(HbaseUtil.listToPut(dataList, rowKey));
        return savePut(puts, tableName);
    }

    private HbaseData parseColumnData(String column, String value) {
        if(StringUtils.isBlank(column) || StringUtils.isBlank(value) ) {
            return null;
        }
        String[] arr = column.split(":");
        if(arr.length != 2) {
            return null;
        }
        HbaseData hbaseData = new HbaseData();
        hbaseData.setFamily(arr[0]);
        hbaseData.setQualifier(arr[1]);
        hbaseData.setValue(value);
        return hbaseData;
    }

    private Table getHbaseTable(String tableName) throws IOException {
        return hbaseConnection.getTable(getHbaseTableName(tableName));
    }

    private Table getHbaseTable(Class<?> clazz) throws IOException {
        return getHbaseTable(getTableName(clazz));
    }

    private  boolean savePut(List<Put> puts, String tableName){
        try(Table table = getHbaseTable(tableName)) {
            //AsyncProcess
            table.put(puts);
            return true;
        } catch (IOException e) {
            String errMsg = e.getMessage();
            if(errMsg.contains("TableNotFoundException")) {


            } else if(errMsg.contains("NoSuchColumnFamilyException")) {
                //family创建好再增加新的family就会报此错误

            } else {

            }
            //RetriesExhaustedWithDetailsException
            log.error("savePut_err1:{}", puts, e);
        }
        return false;
    }

    private  boolean savePut(List<Put> puts, Class<?> clazz){
        return savePut(puts, getTableName(clazz));
    }

    @Override
    public <T> T getRecord(Class<T> clazz, String rowKey) {
        Result result = getResult(getTableName(clazz), rowKey);
        if(result == null) {
            return null;
        }
        try {
            return HbaseUtil.resultToObject(result, clazz);
        } catch (Exception e) {
            log.error("getRecord_err:{},{}", getTableName(clazz), rowKey, e);
        }
        return null;
    }


    @SuppressWarnings("unchecked")
    @Override
    public <T> T getRecord(T obj) {
        return  (T) getRecord(obj.getClass(), HbaseUtil.getRowKey(obj));
    }

    @Override
    public <T> List<T> getRecords(Class<T> clazz, List<String> rowKeys) {
        List<T> objList = new ArrayList<T>();
        List<Result> results = getResults(getTableName(clazz), rowKeys);
        if (results.isEmpty()) {
            return objList;
        }
        for (Result result : results) {
            try {
                objList.add(HbaseUtil.resultToObject(result, clazz));
            } catch (Exception e) {
                log.warn("getHbaseData_err:{}", result, e);
            }
        }
        return objList;
    }

    @Override
    public List<HbaseCell> getCells(String tableName, String rowKey) {
        Result result = getResult(tableName, rowKey);
        return HbaseUtil.resultToHbaseCell(result);
    }




    private Result getResult(String tableName, String rowKey) {
        try {
            Table table = hbaseConnection.getTable(getHbaseTableName(tableName));
            return table.get(new Get(Bytes.toBytes(rowKey)));
        } catch (IOException s) {
            //jSocketTimeoutException: callTimeout=10000, callDuration=11590: Failed after attempts=3, exceptions:This server is in the failed servers list
            //RetriesExhaustedException: Failed after attempts=1, exceptions:CallTimeoutException: Call id=1, waitTime=1, operationTimeout=1 expired
            log.error("getResult_err:{},{}", tableName, rowKey, s);
        }
        return null;
    }


    private List<Result> getResults(String tableName, List<String> rowKeys) {
        List<Result> resultList = new ArrayList<>();
        List<Get> gets = rowKeys.stream().filter(key -> key != null).map(key -> new Get(Bytes.toBytes(key))).collect(Collectors.toList());
        try {
            Table table = hbaseConnection.getTable(getHbaseTableName(tableName));
            Result[] results = table.get(gets);
            Collections.addAll(resultList, results);
        } catch (IOException e) {
            log.error("getResults_err:{},{}", tableName, gets, e);
        }
        return resultList;
    }

    @Override
    public  <T> List<T> search(Class<T> clazz, HbaseQuery query){
        List<T> objList = new ArrayList<>();
        ResultScanner resultScanner = null;
        try (Table table = getHbaseTable(clazz)) {
            resultScanner = table.getScanner(query.getScan());
            for (Result result : resultScanner) {
                objList.add(HbaseUtil.resultToObject(result, clazz));
            }
        } catch (Exception e){
            log.error("queryRowKey_err:{}", query, e);
        } finally {
            if(resultScanner != null){
                try {
                    resultScanner.close();
                } catch (Exception e) {
                    log.error("scanner_close_err", e);
                }
            }
        }
        return objList;
    }

    @Override
    public List<HbaseRow> search(String tableName, HbaseQuery query) {
        ResultScanner resultScanner = null;
        List<HbaseRow> rowList = new ArrayList<>();
        try (Table table = getHbaseTable(tableName)) {
            if(query.getRowKey() != null) {
                Result  result = getResult(tableName, query.getRowKey());
                List<HbaseCell> cells = HbaseUtil.resultToHbaseCell(result);
                if(cells.size() == 0) {
                    return rowList;
                }
                HbaseRow row = new HbaseRow();
                row.setCellList(cells);
                row.setRowKey(query.getRowKey());
                rowList.add(row);
                return rowList;
            }
            resultScanner = table.getScanner(query.getScan());
            int aa=0;
            for (Result result : resultScanner) {
                List<HbaseCell> cells = HbaseUtil.resultToHbaseCell(result);
                HbaseRow row = new HbaseRow();
                row.setCellList(cells);
                row.setRowKey(new String(result.getRow()));
                rowList.add(row);
                if (aa++ ==mymaxrow) break;
            }
        } catch (Exception e){
            log.error("queryRowKey_err:{}", query, e);
        } finally {
            if(resultScanner != null){
                try {
                    resultScanner.close();
                } catch (Exception e) {
                    log.error("scanner_close_err", e);
                }
            }
        }
        return rowList;
    }

    private void close() {
        if(hbaseConnection == null) {
            return;
        }
        try {
            hbaseConnection.close();
            log.info("hbase_closed");
        } catch (Exception e) {
            log.error("hbase_close_err", e);
        }
    }

    private boolean isClosed() {
        return (hbaseConnection == null || hbaseConnection.isClosed());
    }

    private Admin getAdmin() throws IOException {
        if(isClosed()) {
            return null;
        }
        return hbaseConnection.getAdmin();
    }



}
